// Tencent is pleased to support the open source community by making
// 蓝鲸智云 - 监控平台/日志平台 (BlueKing - Monitor/Log) available.
// Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. All rights reserved.
// Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
// You may obtain a copy of the License at http://opensource.org/licenses/MIT
// Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
// an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
// specific language governing permissions and limitations under the License.
//

package outputgseds

import (
	"bytes"
	"crypto/tls"
	"crypto/x509"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"reflect"
	"strconv"
	"time"

	"github.com/TencentBlueKing/bkmonitor-kits/logger"
	"github.com/elastic/beats/libbeat/publisher"
)

const (
	lenDynDataExtHead = 12
)

type PubHead struct {
	ExtType    uint16
	ExtVersion uint16
	MagicNum   uint32
	BodyLen    uint32
}

func NewPubHead(dataLen int) PubHead {
	return PubHead{
		ExtType:    0,
		ExtVersion: 1,
		MagicNum:   0xdeadbeef,
		BodyLen:    uint32(lenDynDataExtHead + dataLen),
	}
}

func (h PubHead) Bytes() []byte {
	buf := &bytes.Buffer{}
	binary.Write(buf, binary.BigEndian, h.ExtType)
	binary.Write(buf, binary.BigEndian, h.ExtVersion)
	binary.Write(buf, binary.BigEndian, h.MagicNum)
	binary.Write(buf, binary.BigEndian, h.BodyLen)
	return buf.Bytes()
}

type DynDataExtHead struct {
	ChannelID  uint32
	Timestamp  uint32
	ExtBodyLen uint32
}

func NewDynDataExtHead(channelID uint32) DynDataExtHead {
	return DynDataExtHead{
		ChannelID:  channelID,
		Timestamp:  uint32(time.Now().Unix()),
		ExtBodyLen: 0, // 无额外的 LvHead
	}
}

func (h DynDataExtHead) Bytes() []byte {
	buf := &bytes.Buffer{}
	binary.Write(buf, binary.BigEndian, h.ChannelID)
	binary.Write(buf, binary.BigEndian, h.Timestamp)
	binary.Write(buf, binary.BigEndian, h.ExtBodyLen)
	return buf.Bytes()
}

type DataHead struct {
	PubHead        PubHead
	DynDataExtHead DynDataExtHead
}

func (h DataHead) Bytes() []byte {
	var bs []byte
	bs = append(bs, h.PubHead.Bytes()...)
	bs = append(bs, h.DynDataExtHead.Bytes()...)
	return bs
}

type Config struct {
	Endpoint     string        `config:"endpoint"`
	WriteTimeout time.Duration `config:"write_timeout"`
	RetryTimes   int           `config:"retry_times"`
	CertFile     string        `config:"cert_file"`
	KeyFile      string        `config:"key_file"`
	CaFile       string        `config:"ca_file"`
}

func (c *Config) Validate() error {
	if c.Endpoint == "" {
		return errors.New("invalid gse_ds output endpoint")
	}
	if c.CertFile == "" {
		return errors.New("invalid gse_ds output cert_file")
	}
	if c.KeyFile == "" {
		return errors.New("invalid gse_ds output key_file")
	}
	if c.CaFile == "" {
		return errors.New("invalid gse_ds output ca_file")
	}

	if c.RetryTimes <= 0 {
		c.RetryTimes = 3
	}
	if c.WriteTimeout <= 0 {
		c.WriteTimeout = time.Second * 30
	}
	return nil
}

type Client struct {
	conn   *tls.Conn
	config *Config
	ch     chan struct{}
}

func NewClient(cfg *Config) (*Client, error) {
	if err := cfg.Validate(); err != nil {
		return nil, err
	}

	client := &Client{config: cfg, ch: make(chan struct{}, 1)}
	if err := client.connect(); err != nil {
		panic(err)
	}

	return client, nil
}

func (c *Client) Start() error { return c.connect() }
func (c *Client) Close() error { return c.conn.Close() }

func (c *Client) connect() error {
	select {
	case c.ch <- struct{}{}:
	default:
		return nil
	}
	defer func() { <-c.ch }()

	cliCert, err := tls.LoadX509KeyPair(c.config.CertFile, c.config.KeyFile)
	if err != nil {
		return err
	}

	caCrt, err := ioutil.ReadFile(c.config.CaFile)
	if err != nil {
		return err
	}

	caCertPool := x509.NewCertPool()
	caCertPool.AppendCertsFromPEM(caCrt)
	tlsConfig := &tls.Config{
		Certificates:       []tls.Certificate{cliCert},
		RootCAs:            caCertPool, // 校验服务端证书的 CA 池
		InsecureSkipVerify: true,
	}

	tlsConfig.BuildNameToCertificate()
	conn, err := tls.Dial("tcp", c.config.Endpoint, tlsConfig)
	if err != nil {
		return err
	}

	c.conn = conn
	return nil
}

func (c *Client) write(b []byte) (int, error) {
	if c.config.WriteTimeout > 0 {
		err := c.conn.SetWriteDeadline(time.Now().Add(c.config.WriteTimeout))
		if err != nil {
			return -1, err
		}
	}
	return c.conn.Write(b)
}

func (c *Client) Send(event publisher.Event) error {
	content := event.Content
	msg := content.Fields
	dataID, err := msg.GetValue("dataid")
	if err != nil {
		return err
	}

	dataid, err := tryFromInt32(dataID)
	if err != nil {
		return err
	}

	msgs, err := json.Marshal(msg)
	if err != nil {
		return err
	}

	dataHead := DataHead{
		PubHead:        NewPubHead(len(msgs)),
		DynDataExtHead: NewDynDataExtHead(uint32(dataid)),
	}

	var bs []byte
	bs = append(bs, dataHead.Bytes()...)
	bs = append(bs, msgs...)

	var count int
loop:
	for count <= c.config.RetryTimes {
		count++

		_, err := c.write(bs)
		if err == nil {
			break loop
		}

		// TODO(mando): 错误即重连是否合理？
		if err := c.connect(); err != nil {
			logger.Errorf("failed to reconnect data server: %v", err)
		}
	}

	return err
}

func tryFromInt32(a interface{}) (int32, error) {
	switch a.(type) {
	case int, int8, int16, int32, int64:
		return int32(reflect.ValueOf(a).Int()), nil
	case uint, uint8, uint16, uint32, uint64:
		return int32(reflect.ValueOf(a).Uint()), nil
	case string:
		i, err := strconv.ParseInt(a.(string), 10, 32)
		return int32(i), err
	default:
		return 0, fmt.Errorf("unsupported type: %v", a)
	}
}
